package com.pie4cloud.pie.job.executor.kaf;

import com.alibaba.fastjson.JSON;
import com.pie4cloud.pie.bus.api.consumer.IKfkaConsumer;
import com.pie4cloud.pie.bus.api.consumer.annotation.KfkConsumer;
import com.pie4cloud.pie.common.core.constant.SecurityConstants;
import com.pie4cloud.pie.common.core.util.R;
import com.pie4cloud.pie.job.api.entity.DispenseCongfigEntity;
import com.pie4cloud.pie.job.api.entity.GatherMessageEntity;
import com.pie4cloud.pie.job.api.entity.PlaybillEntity;
import com.pie4cloud.pie.job.api.feign.RemoteDispenseCongfigService;
import com.pie4cloud.pie.job.api.vo.TimeExpressionVo;
import com.pie4cloud.pie.job.api.vo.TimeLevelVo;
import com.pie4cloud.pie.job.api.vo.UploadVo;
import com.pie4cloud.pie.job.executor.service.DispenseToolService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.io.File;
import java.util.Date;
import java.util.List;

@Slf4j
@Service
@KfkConsumer(handler = "dispenseRealTimeKafkaHandler")
@RequiredArgsConstructor
public class DispenseRealTimeKafkaHandler implements IKfkaConsumer {
	private final RemoteDispenseCongfigService remoteDispenseCongfigService;
	private final DispenseToolService dispenseToolService;
	@Override
	public R consume(String data, String topic, String groupId) throws Exception {
        String id=groupId.replaceAll("DISPENSE_","");
		GatherMessageEntity gatherMessageEntity = JSON.parseObject(data, GatherMessageEntity.class);
		R<DispenseCongfigEntity> r=remoteDispenseCongfigService.getRemoteById(Long.parseLong(id), SecurityConstants.FROM_IN);
		if(r.getCode()==0){
			DispenseCongfigEntity dispenseCongfigEntity=r.getData();
			String filePath=gatherMessageEntity.getLocalFilepath().replaceAll("\\\\","/");;
			File file=new File(filePath);
			TimeLevelVo timeLevelVo=new TimeLevelVo();
			timeLevelVo.setTimeKey(System.currentTimeMillis());
			TimeExpressionVo timeExpressionVo=new TimeExpressionVo();
			dispenseToolService.drawTimeRealExpression(dispenseCongfigEntity.getSourceFileName(),timeExpressionVo);
			List<String> regulars=dispenseToolService.drawRegular(filePath,timeExpressionVo.getSourceFileName());

			UploadVo uploadVo=new UploadVo();

			dispenseToolService.getDateTime(regulars,timeLevelVo,timeExpressionVo,uploadVo);
			R<PlaybillEntity> playbillEntityR=remoteDispenseCongfigService.findByDdataId(dispenseCongfigEntity.getPlayBillId(), SecurityConstants.FROM_IN);
			PlaybillEntity playbillEntity=playbillEntityR.getData();
			Date lastTime=dispenseToolService.getLastTriggerTime(playbillEntity.getJobCron(),new Date(uploadVo.getDdataTime()));
			Date nextTime=dispenseToolService.getNextTriggerTime(playbillEntity.getJobCron(),lastTime);
			timeLevelVo.setRealTime(true);
			timeLevelVo.setStartTime(lastTime.getTime());
			timeLevelVo.setEndTime(nextTime.getTime());
			timeLevelVo.setTimeKey(lastTime.getTime());
			dispenseToolService.rulesDirectoryAndFileName(dispenseCongfigEntity,file,timeLevelVo,timeExpressionVo,uploadVo);
		}
		return R.ok();
	}
}
